/******************************************************************************
 *  author: enquanyan
 *  version: 1.0
 *  description: forward_conn implementation
 *  date: 2014-08-04
 *  other: 无
 ******************************************************************************/

#include "forward_conn.h"

static const char *boundary = "---------------------------7da2137580612";

static const char *end_boundary = "\r\n-----------------------------7da2137580612--\r\n";
        
static const char* const content_start = "-----------------------------7da2137580612\r\nContent-Disposition:form-data; name=\"filename\"; filename=";
        
static const char* const content_type = "Content-Type: application/octet-stream\r\n\r\n";

static const char* const content_type_header ="Content-Type:multipart/form-data;boundary=---------------------------7da2137580612";

int addfd_EPOLLOUT(int epollfd, abstract_conn *abstract_conn_ptr, int fd, bool one_shot)
{
	epoll_event event;
    abstract_conn_ptr->m_sockfd = fd;
    
	event.data.ptr = (void*)abstract_conn_ptr;
    
	event.events = EPOLLOUT | EPOLLET | EPOLLRDHUP;
    
	if (one_shot)
	{
		event.events |= EPOLLONESHOT;
	}

	if (epoll_ctl(epollfd, EPOLL_CTL_ADD, fd, &event) < 0)
	{
        pr_debug("EPOLLOUT EPOLL_CTL_ADD error->sockfd:%d\n", fd);
		return -1;
	}

	return setnonblocking(fd);
}

int forward_conn::m_epollfd = -1;

forward_conn::forward_conn()
{
    this->conn_type = FORWARD_CONN_T;
}
        
forward_conn::~forward_conn()
{
            //do nothing
}

void forward_conn::close_conn(bool real_close) 
{
	if (real_close && (m_sockfd != -1))
	{
		
		pr_debug("forward conn close_conn->sockfd:%d->thread_no:%d\n", m_sockfd, thread_no);
		
		removefd(m_epollfd, m_sockfd, is_in_epoll);
		
        /*
		thread_arg *thread_arg_pointer = threadpool<abstract_conn>::thread_arg_array + thread_no;
		
		thread_arg_pointer->active_http_conn--;
		
		int active_http_conn = thread_arg_pointer->active_http_conn.load(std::memory_order_seq_cst);
		
		pr_debug("close_conn->sockfd:%d->thread_no:%d->--active_forward_conn:%d\n", m_sockfd, thread_no, active_http_conn);
		
		if(thread_arg_pointer->is_first)
		{
			thread_arg_pointer->interval_handled_http_conn_first++;
		}
		else
		{
			thread_arg_pointer->interval_handled_http_conn_second++;
		}
        */
		
	}
    
    if(send_file_id > 0)
    {
        close(send_file_id);
        send_file_id = -1;
    }
    
    delete this;
}

bool forward_conn::init(int remote_port, char* remote_ip) 
{
    this->remote_port = remote_port;
    strcpy(this->remote_ip, remote_ip);
    
	if(!connect_remote())
    {
        pr_debug("connect remote error->errno:%d\n", errno);
        
        return false;
    }
    
    init();
    
    m_real_file[0] = '\0';
    strcat(m_real_file, doc_root);
    strcat(m_real_file, "/");
    strcat(m_real_file, m_upload_file_name);
    
    generate_request();
	
	int error = 0;
	socklen_t len = sizeof(error);
	getsockopt(m_sockfd, SOL_SOCKET, SO_ERROR, &error, &len);
	
	int reuse = 1;
	setsockopt(m_sockfd, SOL_SOCKET, SO_REUSEADDR, &reuse, sizeof(reuse));
    
    return true;
}

void forward_conn::init() 
{
	m_check_state = PARSE_RESPONSELINE_STATE;
	m_linger = false;

	m_content_length = 0;
	m_host = NULL;
    
	m_start_line = 0;
	m_checked_idx = 0;
	m_read_idx = 0;
	m_write_idx = 0;
    
	send_count = 0;
	send_file_count = 0;
    send_end_count = 0;
    
    send_file_id = -1;
	
	is_in_epoll = false;
    
    end_size = strlen(end_boundary);

	//read_handle = &forward_conn::read_request_line_and_headers;
	//write_handle = &forward_conn::write_response;
}

LINE_STATUS forward_conn::parse_line() 
{
	char temp;
	
	for (; m_checked_idx < m_read_idx; ++m_checked_idx) 
	{
		temp = m_read_buf[m_checked_idx];
		if (temp == '\r') 
		{
			if ((m_checked_idx + 1) == m_read_idx)
			{
				return LINE_OPEN;
			} 
			else if (m_read_buf[m_checked_idx + 1] == '\n')
			{
				m_read_buf[m_checked_idx++] = '\0';
				m_read_buf[m_checked_idx++] = '\0';
				return LINE_OK;
			}

			return LINE_BAD;
		} 
		else if (temp == '\n')
		{
			if ((m_checked_idx > 1)
					&& (m_read_buf[m_checked_idx - 1] == '\r'))
			{
				m_read_buf[m_checked_idx - 1] = '\0';
				m_read_buf[m_checked_idx++] = '\0';
				return LINE_OK;
			}
			
			return LINE_BAD;
			
		}
	}

	return LINE_OPEN;
}

READ_STATUS forward_conn::read_response_line_and_headers() 
{
	//pr_debug("read_request_line_and_headers->sockfd:%d\n", m_sockfd);
	
	if (m_read_idx >= READ_BUFFER_SIZE - 1)
	{
		return BUFF_IS_FULL;
	}

	int bytes_read = 0;
	
	while (true) 
	{
		//pr_debug("read_request_line_and_headers->into while->sockfd:%d\n", m_sockfd);
		
		bytes_read = recv(m_sockfd, m_read_buf + m_read_idx,
				READ_BUFFER_SIZE - m_read_idx, 0);

		//pr_debug("read_request_line_and_headers->bytes_read:%d->sockfd:%d\n", bytes_read, m_sockfd);
		
		if (bytes_read == -1) 
		{
			if ( errno == EAGAIN || errno == EWOULDBLOCK) 
			{
				this->is_read_ready = false;
                
				//pr_debug("read_request_line_and_headers->EAGAIN->sockfd:%d\n", m_sockfd);
                
				return READ_OK;
			}
			
			pr_debug("read_request_line_and_headers->recv(-1)->errno:%d->sockfd:%d\n", errno, m_sockfd);
			
			return READ_ERROR;
		}
		else if (bytes_read == 0)
		{
			pr_debug("read_request_line_and_headers->recv(0)->errno:%d->sockfd:%d\n", errno, m_sockfd);
			
			return PEER_CLOSED;
		}
		
        /*
		//statistic data
		thread_arg *thread_arg_pointer = threadpool<abstract_conn>::thread_arg_array + thread_no;	
        	
		if(thread_arg_pointer->is_first)
		{
			thread_arg_pointer->interval_received_bytes_first += bytes_read;
		}
		else
		{
			thread_arg_pointer->interval_received_bytes_second += bytes_read;
		}
        */

		m_read_idx += bytes_read;
		
		if (m_read_idx >= READ_BUFFER_SIZE - 1) 
		{
			
			return BUFF_IS_FULL;
		}
		
	}
	return READ_OK;
}

HTTP_CODE forward_conn::parse_response_line(char* text)
{
	m_url = strpbrk(text, " \t");
	
	if (m_url == NULL) 
	{
		return BAD_REQUEST;
	}
	
	*m_url++ = '\0';

	m_version = text;
    
    if (strcasecmp(m_version, "HTTP/1.1") != 0 && strcasecmp(m_version, "HTTP/1.0") != 0) 
	{
		return BAD_REQUEST;
	}

	m_url += strspn(m_url, " \t");
    m_status_code = m_url;
    
    m_url = strpbrk(m_url, " \t");
    
	if (m_url == NULL) 
	{
		return BAD_REQUEST;
	}
    
    *m_url++ = '\0';
    m_response_str = m_url;

	return GO_TO_NEXT;
}

HTTP_CODE forward_conn::parse_headers(char* text)
{
	
	if (text[0] == '\0') 
	{

		if (m_content_length != 0) 
		{

			return GO_TO_NEXT;
		}

		return HANDLE_REQUEST;

	} 
	else if (strncasecmp(text, "Connection:", 11) == 0) 
	{

		text += 11;
		text += strspn(text, " \t");
		if (strcasecmp(text, "keep-alive") == 0) {
			//m_linger = true;
		}

	} 
	else if (strncasecmp(text, "Content-Length:", 15) == 0) 
	{

		text += 15;
		text += strspn(text, " \t");
		m_content_length = atol(text);

	} 
	else if (strncasecmp(text, "Host:", 5) == 0) 
	{

		text += 5;
		text += strspn(text, " \t");
		m_host = text;

	} 
	else {

		//pr_debug("oop! unknow header %s\n", text);

	}

	return UNCOMPLETE_REQUEST;
}

HTTP_CODE forward_conn::parse_response_content(char* text) 
{
    pr_debug("parse response content before before handle->m_sockfd:%d\n", m_sockfd);
    
	if (m_read_idx >= (m_content_length + m_checked_idx)) 
	{
        pr_debug("parse response content before handle->m_sockfd:%d\n", m_sockfd);
		text[m_content_length] = '\0';
		return HANDLE_REQUEST;
	}

	return UNCOMPLETE_REQUEST;
}

void forward_conn::unmap() 
{
	/*if (m_file_address) 
	{
		munmap(m_file_address, m_file_stat.st_size);
		m_file_address = 0;
	}*/
    
    if(send_file_id > 0)
    {
        close(send_file_id);
        send_file_id = -1;
    }
}

bool forward_conn::add_request(const char* format, ...) 
{
	if (m_write_idx >= WRITE_BUFFER_SIZE)
	{
		return false;
	}
	
	va_list arg_list;
	va_start(arg_list, format);
	
	int len = vsnprintf(m_write_buf + m_write_idx,
			WRITE_BUFFER_SIZE - 1 - m_write_idx, format, arg_list);
			
	if (len >= (WRITE_BUFFER_SIZE - 1 - m_write_idx)) 
	{
		return false;
	}
	
	m_write_idx += len;
	
	va_end(arg_list);
	
	return true;
}

bool forward_conn::add_request_line()
{
	return add_request("%s %s %s\r\n", "POST", FORWARD_STRING, "HTTP/1.1");
}

bool forward_conn::add_headers(int content_len) 
{
	add_content_length(content_len);
	add_linger();
    add_content_type();
	add_blank_line();
}

bool forward_conn::add_content_length(int content_len) 
{
	return add_request("Content-Length: %d\r\n", content_len);
}

bool forward_conn::add_content_type() 
{
	return add_request("%s\r\n", content_type_header);
}

bool forward_conn::add_linger() 
{
	return add_request("Connection: %s\r\n",
			(m_linger == true) ? "keep-alive" : "close");
}

bool forward_conn::add_blank_line() 
{
	return add_request("%s", "\r\n");
}

bool forward_conn::add_content(const char* content) 
{
	return add_request("%s", content);
}

READ_STATUS forward_conn::read_response() 
{
    return read_response_line_and_headers();
}

HTTP_CODE forward_conn::parse_response_line() 
{
	LINE_STATUS line_status = LINE_OK;
	HTTP_CODE ret = UNCOMPLETE_REQUEST;
	char* text = 0;

	line_status = parse_line();
	
	if (line_status == LINE_OK) 
	{
		text = get_line();
		m_start_line = m_checked_idx;
		return parse_response_line(text);
	} 
	else if (line_status == LINE_BAD)
	{
		return BAD_REQUEST;
	}
	
	return UNCOMPLETE_REQUEST;
}

HTTP_CODE forward_conn::parse_headers() 
{    
	LINE_STATUS line_status = LINE_OK;
	HTTP_CODE ret = UNCOMPLETE_REQUEST;
	char* text = 0;

	while (true) 
	{
		line_status = parse_line();
		
		switch (line_status) 
		{
            case LINE_OK:
                text = get_line();
                m_start_line = m_checked_idx;
                
                ret = parse_headers(text);
                
                //pr_debug("%d:ret\n", ret);
                
                if (ret == GO_TO_NEXT) 
                {
                    return GO_TO_NEXT;
                }
                else if (ret == BAD_REQUEST) 
                {
                    pr_debug("parse response header error\n");
                    
                    return BAD_REQUEST;
                } 
                else if (ret == HANDLE_REQUEST)
                {
                    return HANDLE_REQUEST;
                }
                break;
                
            case LINE_BAD:
                pr_debug("parse response header error\n");
                
                return BAD_REQUEST;
                
            default:
                return UNCOMPLETE_REQUEST;
		}
	}
}

HTTP_CODE forward_conn::parse_response_content() 
{
    pr_debug("parse response content->m_sockfd:%d\n", m_sockfd);
	LINE_STATUS line_status = LINE_OK;
	HTTP_CODE ret = UNCOMPLETE_REQUEST;
	char* text = 0;

    pr_debug("before parse line->m_sockfd:%d\n", m_sockfd);
	//line_status = parse_line();
	pr_debug("after parse line->m_sockfd:%d\n", m_sockfd);
	if (line_status == LINE_OK) 
	{
        pr_debug("LINE_OK before get_line->m_sockfd:%d\n", m_sockfd);
		text = get_line();
		m_start_line = m_checked_idx;
		return parse_response_content(text);
	} 
	else if (line_status == LINE_BAD)
	{
		return BAD_REQUEST;
	}
	
	return UNCOMPLETE_REQUEST;
}

HTTP_CODE forward_conn::parse_response() 
{
	//pr_debug("parse_request->sockfd:%d\n",m_sockfd);
	
	HTTP_CODE http_code;
	
	switch (m_check_state) 
	{
		case PARSE_RESPONSELINE_STATE:
		
			pr_debug("before parse response line state->sockfd:%d\n", m_sockfd);
			
			http_code = parse_response_line();
			
			if (http_code == BAD_REQUEST) 
			{
				pr_debug("parse response line error\n");
				
				return BAD_REQUEST;
			} 
			else if (http_code == UNCOMPLETE_REQUEST) 
			{
				return UNCOMPLETE_REQUEST;
			} 
			else if (http_code == GO_TO_NEXT) 
			{
				m_check_state = PARSE_RESPONSE_HEADER_STATE;
			} 
			else 
			{
				return http_code;
			}
			
		case PARSE_RESPONSE_HEADER_STATE:
			
            pr_debug("before parse response header state->sockfd:%d\n", m_sockfd);
            
			http_code = parse_headers();
			
			if (http_code == BAD_REQUEST) 
			{
				
				pr_debug("parse response headers error\n");
				
				return BAD_REQUEST;
			} 
			else if (http_code == UNCOMPLETE_REQUEST)
			{
				return UNCOMPLETE_REQUEST;
			} 
			else if (http_code == GO_TO_NEXT)
			{
                m_check_state = PARSE_RESPONSE_CONTENT_STATE;
			} 
			else 
			{
				return http_code;
			}

		case PARSE_RESPONSE_CONTENT_STATE:
			
            pr_debug("before parse response content state->sockfd:%d\n", m_sockfd);
            	
			http_code = parse_response_content();
            			
			if (http_code == BAD_REQUEST) 
			{
				return BAD_REQUEST;
			} 
			else if (http_code == UNCOMPLETE_REQUEST) 
			{
				return UNCOMPLETE_REQUEST;
			} 
			else
			{
				return http_code;
			}

		default:
			return INTERNAL_ERROR;
	}
	
}

HTTP_CODE forward_conn::read_handle_func() 
{
	READ_STATUS read_status;
	
	while (this->is_read_ready)
	{
		//pr_debug("is_read_ready\n");
		
		read_status = read_response();
		
		//pr_debug("read_request return->sockfd:%d\n", m_sockfd);
        pr_debug("read_buffer:%s->sockfd:%d\n", m_read_buf,m_sockfd);
		
		switch (read_status) 
		{
			case READ_OK:
			case BUFF_IS_FULL:
				switch (parse_response()) 
				{
					case HANDLE_REQUEST:
						close_conn();
						return HANDLE_REQUEST;
						
					case BAD_REQUEST:
						close_conn();
						return BAD_REQUEST;
						
					case FILE_EXIST_REQUEST:
						close_conn();
						return FILE_EXIST_REQUEST;
						
					case UNCOMPLETE_REQUEST:
						if (read_status != BUFF_IS_FULL) 
						{
							modfd(m_epollfd, this, m_sockfd, EPOLLIN, is_in_epoll);
							is_in_epoll = true;
							
							return UNCOMPLETE_REQUEST;
						}
						
						break;
                    default:
                        close_conn();
                        return ERROR_REQUEST;
				 }
					
					break;

				case PEER_CLOSED:

				case READ_ERROR:
					pr_debug("read_handle_func->READ_ERROR->sockfd:%d\n",m_sockfd);
					
					close_conn();
					
					return ERROR_REQUEST;
		}
	}

	if (!is_read_ready) 
	{
		modfd(m_epollfd, this, m_sockfd, EPOLLIN, is_in_epoll);
		is_in_epoll = true;
		return UNCOMPLETE_REQUEST;
	}
}

HTTP_CODE forward_conn::write_handle_func() 
{
	send_request();
	return RESPONSE_ERROR;
}

void forward_conn::process() 
{    
	switch (handle_type) 
    {
        case READ:
            read_handle_func();
            break;
            
        case WRITE:
            write_handle_func();
            break;

        default:
            //do nothing
            break;
	}
}

bool forward_conn::generate_request()
{
    if (stat(m_real_file, &m_file_stat) < 0)
	{
		pr_debug("NO_RESOURCE\n");
		return false;
	}
    
    m_content_length = strlen(content_start) + strlen(content_type) + strlen(m_upload_file_name) + 4 + strlen(end_boundary) + m_file_stat.st_size;
    
    send_file_id = open(m_real_file, O_RDONLY);
    
	//m_file_address = (char*) mmap(0, m_file_stat.st_size, PROT_READ, MAP_PRIVATE, send_file_id, 0);
    
    send_file_size = m_file_stat.st_size;
    
	//close(send_file_id);
    
    add_request_line();
    add_headers(m_content_length);
    
    add_request("%s",content_start);
    add_request("\"%s\"\r\n",m_upload_file_name);
    add_request("%s",content_type);
    
    pr_debug("m_wirte_buf:%s\n", m_write_buf);
    
}

bool forward_conn::send_request() 
{
	int temp = 0;
	
	while (true)
	{
		if (send_count >= m_write_idx) 
		{
			break;
		}
		
		temp = write(m_sockfd, m_write_buf + send_count,
				m_write_idx - send_count);
				
		if (temp <= -1)
		{
			//pr_debug("download sockfd busy: %d\n", errno);
			
			if ( errno == EAGAIN || errno == EWOULDBLOCK)
			{
				
				modfd(m_epollfd, this, m_sockfd, EPOLLOUT, is_in_epoll);
                is_in_epoll = true;
				return true;
			}
			
			pr_debug("write response header error errno:%d\tm_sockfd:%d\n", errno,m_sockfd);
			
            unmap();
			close_conn();
			return false;
		}

		send_count += temp;
		
		if (send_count >= m_write_idx) 
		{
			//pr_debug("write response->sockfd:%d\n", m_sockfd);
			
			break;
		}
	}

	//if (m_file_address != NULL)
    if (send_file_id > 0)
	{
		//pr_debug("start write file\n");
		
		while (true) 
		{
			if (send_file_count >= send_file_size)
			{
				unmap();
				break;
			}
			
			//temp = write(m_sockfd, m_file_address + send_file_count, send_file_size - send_file_count);
            
            temp = sendfile(m_sockfd, send_file_id, &send_file_count, send_file_size - send_file_count);
					
			if (temp <= -1) 
			{
				if ( errno == EAGAIN || errno == EWOULDBLOCK) 
				{
					modfd(m_epollfd, this, m_sockfd, EPOLLOUT, is_in_epoll);
                    is_in_epoll = true;
					return true;
				}
				
				pr_debug("write file error:%d->sockfd:%d\n", errno, m_sockfd);
				
				unmap();
				close_conn();
				return false;
			}

			//send_file_count += temp;
			
			if (send_file_count >= send_file_size)
			{
				unmap();
				break;
			}
		}
	}
    
    temp = 0;
    
    while (true)
	{
		if (send_end_count >= end_size) 
		{
			break;
		}
		
		temp = write(m_sockfd, end_boundary + send_end_count,
				end_size - send_end_count);
				
		if (temp <= -1)
		{
			//pr_debug("download sockfd busy: %d\n", errno);
			
			if ( errno == EAGAIN || errno == EWOULDBLOCK)
			{
				modfd(m_epollfd, this, m_sockfd, EPOLLOUT, is_in_epoll);
                is_in_epoll = true;
				return true;
			}
			
			pr_debug("send request error errno:%d\tm_sockfd:%d\n", errno,m_sockfd);
			
			close_conn();
			return false;
		}

		send_end_count += temp;
		
		if (send_end_count >= end_size) 
		{
			//pr_debug("write response->sockfd:%d\n", m_sockfd);
			break;
		}
	}
	
    pr_debug("forward conn EPOLLIN->sockfd:%d\n", m_sockfd);
	modfd(m_epollfd, this, m_sockfd, EPOLLIN, is_in_epoll);
    is_in_epoll = true;
    
	return true;
}

bool forward_conn::connect_remote()
{
	struct sockaddr_in server_addr, client_addr;
	socklen_t socklen = sizeof(server_addr);

	if((m_sockfd = socket(AF_INET, SOCK_STREAM, 0)) < 0)
	{
		pr_debug("create socket error->errno:%d\n", errno);
        return false;
	}
    
	bzero(&client_addr, sizeof(client_addr));
	client_addr.sin_family = AF_INET;
	client_addr.sin_addr.s_addr = htons(INADDR_ANY);

	bzero(&server_addr, sizeof(server_addr));
	server_addr.sin_family = AF_INET;
	inet_aton(remote_ip, &server_addr.sin_addr);
	server_addr.sin_port = htons(remote_port);

	if(connect(m_sockfd, (struct sockaddr*)&server_addr, socklen) < 0)
	{
		pr_debug("connect error->errno:%d\n", errno);
		return false;
	}
    
    return true;
}

